package com.interauto.service.kaklis;


import com.alibaba.fastjson.JSON;
import com.interauto.Dao.entity.Runtasktable;
import com.interauto.bean.AutoCaseBean;
import com.interauto.service.AutoCaseBeanService;
import com.interauto.service.AutoRunCaseinfoService;
import com.interauto.service.alarm.AlarmSendService;
import com.sun.xml.internal.ws.api.model.MEP;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.Locale;

import static com.interauto.util.AutoStringUtils.isNumeric;

/**
 * 监听和处理任务相关的kafka相关
 */
@Component
public class RunTaskConsumer {
    //private static final String runtasktopic="autoruntaskinfo";
    private static final Logger logger= LoggerFactory.getLogger(RunTaskConsumer.class);

    @Autowired
    private AlarmSendService alarmSendService;
    @Autowired
    private AutoCaseBeanService autoCaseBeanService;
    @Autowired
    private AutoRunCaseinfoService autoRunCaseinfoService;

    @KafkaListener(topics = "autoruntaskinfo",groupId = "auto_consumer_group")
    public void taskconsumer(String taskinfo){
        logger.info(String.format("-->>>taskconsumer is:%s",taskinfo));
        if(StringUtils.isEmpty(taskinfo)){
            return;
        }
        try{
            Runtasktable runtasktable= JSON.parseObject(taskinfo,Runtasktable.class);
            String caseinfo=runtasktable.getCaseinfo();
            String callnumber=runtasktable.getCallnumber();
            if(StringUtils.isEmpty(caseinfo) || StringUtils.isEmpty(callnumber)){
                logger.info("info is empty;task is-->"+taskinfo);
                return;
            }
            String[] casearrays=caseinfo.split(",");
            if(casearrays.length<1){
                logger.info("casearrays is empty;task is-->"+taskinfo);
                return;
            }
            for(String caseid:casearrays){
                if(!isNumeric(caseid)){
                    logger.info("caseid is fail;caseid is-->"+caseid);
                    continue;
                }
                AutoCaseBean autoCaseBean=autoCaseBeanService.getAutoCaseinfo(Integer.valueOf(caseid));
                if(autoCaseBean == null){
                    logger.info("getAutoCaseinfo is null; caseid is:"+caseid);
                    continue;
                }
                logger.info("run case info caseid is:"+caseid);
                autoRunCaseinfoService.runcaseinfo(autoCaseBean,runtasktable);
            }
        }
        catch (Exception e){
            logger.error("taskconsumer is system error --->"+e);
        }
    }


    @KafkaListener(topics = "autoalarmtopic",groupId = "auto_consumer_group")
    public void autoalarmconsumer(String message){
        logger.info(String.format("-->>>autoalarmconsumer is:%s",message));
        alarmSendService.sendalarmmessage(message);
    }
}
